package routers

import (
	"bufio"
	"database/sql"
	"fmt"
	"github.com/gin-gonic/gin"
	"go-log/config"
	"go-log/db"
	"io"
	"log"
	"net/http"
	"os"
	"strconv"
	"strings"
	"sync"
	"time"
)

type LogLine struct {
	Row  int
	Line string
}

type LogContext struct {
	Ts         string
	Row        int
	Level      string
	TraceId    string
	ThreadName string
	ClassName  string
	LineNumber int
	Context    string
}

var gw sync.WaitGroup

func batchInsert(goroutineId int, bachLog []LogContext, appName string, nodeId string) {
	insertSql := "insert into `go_logs`.`logs`"
	values := make([]string, 0)
	for _, val := range bachLog {
		value := fmt.Sprintf("('%v','%v','%v', %v, '%v','%v','%v','%v',%v,'%v')",
			val.Ts,
			appName,
			nodeId,
			val.Row,
			val.Level,
			val.TraceId,
			val.ThreadName,
			val.ClassName,
			val.LineNumber,
			val.Context,
		)
		values = append(values, value)
	}
	insertSql = fmt.Sprintf("%v VALUES %v", insertSql, strings.Join(values, ", "))
	exec, err := db.TaosDb.Exec(insertSql)
	if err != nil {
		fmt.Println(err)
	}
	affectedRows, _ := exec.RowsAffected()
	fmt.Println(time.Now().Format("2006-01-02 15:04:05.000"), goroutineId, affectedRows)
}

func transferLog(goroutineId int, logLineChan chan LogLine, appName string, nodeName string) {
	batchSize := config.Conf.BatchSize
	batch := make([]LogContext, 0, batchSize)
	for logLine := range logLineChan {
		split := strings.Split(logLine.Line, "|")
		split2 := strings.Split(split[4], ":")
		className := split2[0]
		lineNumber, _ := strconv.Atoi(split2[1])
		logContext := LogContext{
			Ts:         split[0],
			Row:        logLine.Row,
			Level:      split[1],
			TraceId:    split[2],
			ThreadName: split[3],
			ClassName:  className,
			LineNumber: lineNumber,
			Context:    strings.ReplaceAll(strings.ReplaceAll(split[5], `"`, `\"`), `'`, `\'`),
		}
		batch = append(batch, logContext)
		if len(batch) == batchSize {
			batchInsert(goroutineId, batch, appName, nodeName)
			batch = make([]LogContext, 0, batchSize)
		}
	}
	if len(batch) > 0 {
		batchInsert(goroutineId, batch, appName, nodeName)
	}
	gw.Done()
}

func sourceFile(logLineChan chan LogLine, logFilePath string) {
	file, err := os.OpenFile(logFilePath, os.O_RDONLY, 0655)
	if err != nil {
		log.Fatalln("日志文件打开失败:", err)
	}
	defer func(file *os.File) {
		err := file.Close()
		if err != nil {
			fmt.Println("Error close file:", err)
		}
	}(file)
	reader := bufio.NewReader(file)
	row := 0
	for {
		line, err := reader.ReadString('\n')
		if line != "" && !strings.Contains(line, "|") {
			row++
			continue
		}
		if err == io.EOF {
			break
		}
		row++
		logLineChan <- LogLine{row, strings.Trim(line, "\r\n")}
	}
	close(logLineChan)
	gw.Done()
}

func etl() int64 {
	start := time.Now().UnixMilli()

	for _, logMeta := range config.Conf.LogMetas {
		if !logMeta.Enable {
			continue
		}
		var logLineChan = make(chan LogLine, 1024)

		gw.Add(1)
		go sourceFile(logLineChan, logMeta.LogFilePath)

		for i := range config.Conf.GoroutineSize {
			gw.Add(1)
			go transferLog(i+1, logLineChan, logMeta.AppName, logMeta.NodeId)
		}
	}

	gw.Wait()
	speedMs := time.Now().UnixMilli() - start
	log.Printf("用时=>%dms\n", speedMs)
	return speedMs
}

func appendTs(sql *string, hasWhere *bool, c *gin.Context) {
	yesterday := time.Now().Add(-24 * time.Hour)

	startTs := c.DefaultQuery("startTs", time.Date(yesterday.Year(), yesterday.Month(), yesterday.Day(), 0, 0, 0, 0, time.Local).Format("2006-01-02 15:04:05"))
	endTs := c.DefaultQuery("endTs", time.Date(yesterday.Year(), yesterday.Month(), yesterday.Day(), 23, 59, 59, 999, time.Local).Format("2006-01-02 15:04:05"))

	tsSql := fmt.Sprintf("ts >= '%v' AND ts < '%v'", startTs, endTs)
	if !*hasWhere {
		*sql = *sql + " WHERE " + tsSql
		*hasWhere = true
	} else {
		*sql = *sql + " AND " + tsSql
	}
}

func appendTraceId(sql *string, hasWhere *bool, c *gin.Context) {
	traceId := c.Query("traceId")
	if traceId != "" {
		traceIdSql := fmt.Sprintf("trace_id = '%v'", traceId)
		if !*hasWhere {
			*sql = *sql + " WHERE " + traceIdSql
			*hasWhere = true
		} else {
			*sql = *sql + " AND " + traceIdSql
		}
	}
}

func appendNodeId(sql *string, hasWhere *bool, c *gin.Context) {
	nodeId := c.Query("nodeId")
	if nodeId != "" {
		traceIdSql := fmt.Sprintf("node_id = '%v'", nodeId)
		if !*hasWhere {
			*sql = *sql + " WHERE " + traceIdSql
			*hasWhere = true
		} else {
			*sql = *sql + " AND " + traceIdSql
		}
	}
}

func appendContext(sql *string, hasWhere *bool, c *gin.Context) {
	context := c.Query("context")
	matchType := c.DefaultQuery("matchType", "MATCH_ANY")
	if context != "" {
		matchSql := fmt.Sprintf("context %v '%v'", matchType, context)
		if !*hasWhere {
			*sql = *sql + " WHERE " + matchSql
			*hasWhere = true
		} else {
			*sql = *sql + " AND " + matchSql
		}
	}
}

func appendOrder(sql *string, c *gin.Context) {
	orderBy := c.DefaultQuery("orderBy", "0")
	if orderBy == "1" {
		orderBy = "DESC"
	} else {
		orderBy = "ASC"
	}
	*sql += fmt.Sprintf(" ORDER BY file_row_number %v ", orderBy)
}

func appendLimit(sql *string, c *gin.Context) {
	limit := c.DefaultQuery("limit", "1000")
	*sql += fmt.Sprintf(" LIMIT %v ", limit)
}

func queryDb(baseSql *string, logRows *[]string) {
	log.Printf("%v", *baseSql)
	start := time.Now().UnixMilli()
	rows, err := db.TaosDb.Query(*baseSql)
	if err != nil {
		panic(err)
	}
	defer func(rows *sql.Rows) {
		err := rows.Close()
		if err != nil {
			panic(err)
		}
	}(rows)
	speedMs := time.Now().UnixMilli() - start
	log.Printf("doris 检索用时=>%dms\n", speedMs)

	for rows.Next() {
		var (
			ts            []uint8
			level         string
			traceId       string
			threadName    string
			className     string
			context       string
			nodeId        string
			lineNumber    int
			fileRowNumber int
		)
		// 将查询结果扫描到结构体字段
		err := rows.Scan(&ts, &level, &traceId, &threadName, &className, &context, &nodeId, &lineNumber, &fileRowNumber)
		if err != nil {
			panic(err)
		}
		*logRows = append(*logRows, fmt.Sprintf("%v [%v] [%v] [%v] {%v} [%v:%v] %v", string(ts), level, traceId, threadName, nodeId, className, lineNumber, context))
	}
}

func ApiRouter(r *gin.Engine) {
	router := r.Group("/log/")

	// 日志导入
	router.GET("/init", func(c *gin.Context) {
		c.JSON(http.StatusOK, gin.H{
			"code": 0,
			"msg":  "success",
			"data": etl(),
		})
	})

	// 日志检索
	router.GET("/search", func(c *gin.Context) {
		baseSql := "SELECT `ts`,`level`,`trace_id`,`thread_name`,`class_name`,`context`,`node_id`, `line_number`, `file_row_number` FROM `go_logs`.`logs`"
		hasWhere := false
		appendTs(&baseSql, &hasWhere, c)
		appendNodeId(&baseSql, &hasWhere, c)
		appendTraceId(&baseSql, &hasWhere, c)
		appendContext(&baseSql, &hasWhere, c)
		appendOrder(&baseSql, c)
		appendLimit(&baseSql, c)

		var logRows []string
		queryDb(&baseSql, &logRows)

		c.Header("Content-Type", "text/plain; charset=utf-8")
		c.String(http.StatusOK, strings.Join(logRows, "\n"))
	})
}
